%%%-------------------------------------------------------------------
%%% @doc  
%%% @end
%%%-------------------------------------------------------------------
-module(dp_server).

-behaviour(gen_server).
%%--------------------------------------------------------------------
%% Include files
%%--------------------------------------------------------------------
-include_lib("mnemosyne/include/mnemosyne.hrl").
-include("/usr/local/lib/yaws/include/yaws.hrl").
-include("drproxy.hrl").

%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------

-export([start_link/0, register/2, client/0, clients/0, status/0, set/2]).
-export([total_requests/0, requests_per_client/0, restart/1,
		restart_all/0, averages/0, client_queues/1]).

%%--------------------------------------------------------------------
%% gen_server callbacks
%%--------------------------------------------------------------------

-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	terminate/2, code_change/3]).
	
%%--------------------------------------------------------------------
%% record definitions
%%--------------------------------------------------------------------

-record(state, 
	{
		clients = [],
		averages = {0.0, 0.0, 0.0},
		one_minute = {now(), 0},
		five_minutes = {now(), 0},
		fifteen_minutes = {now(), 0}
	}
).

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% @doc Starts the server.
%% @spec start_link() -> {ok, pid()} | {error, Reason}
%% @end
%%--------------------------------------------------------------------
start_link() ->
	gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

total_requests() ->
	gen_server:call(?MODULE, total_requests).

requests_per_client() ->
	gen_server:call(?MODULE, requests_per_client).
	
averages() ->
	gen_server:call(?MODULE, averages).

register(ProcessLoop, Node) ->
    gen_server:call(?MODULE, {register, ProcessLoop, Node}).

client() ->
	receive
		{register, ProcessLoop, Node, Pid} ->
			Response = ?MODULE:register(ProcessLoop, Node),
			Pid ! Response,
			client_loop()
	end.
	
clients() ->
	gen_server:call(?MODULE, clients).

status() ->
	gen_server:cast(?MODULE, status).

set(Node, Weight) ->
	gen_server:cast(?MODULE, {set, Node, Weight}).

restart(Node) ->
	gen_server:cast(?MODULE, {restart, Node}).

restart_all() ->
	gen_server:cast(?MODULE, restart_all).

client_queues(ClientQueues) ->
  gen_server:cast(?MODULE, {client_queues, ClientQueues}).

%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% Function: init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([]) ->
	process_flag(trap_exit, true),
	start_server(),
	start_yaws(),
	change_logfile(),
	{ok, #state{}}.

%%--------------------------------------------------------------------
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call(total_requests, _From, State) ->
	{reply, lists:foldl(fun({_Node, _Client, _Weight, R, _Q}, Total) -> Total + R end, 0, State#state.clients), State};

handle_call(requests_per_client, _From, State) ->
	{reply, [lists:map(fun({Node, _Client, Weight, R, _Q}) -> {Node, Weight, R} end, State#state.clients)], State};

handle_call(averages, _From, State) ->
	NewState = log_request(0, State, State#state.clients),
	{reply, NewState#state.averages, NewState};

handle_call({update_clients, NewClients}, _From, State) ->
	{noreply, State#state{clients = NewClients}};

handle_call(clients, _From, State) ->
	{reply, State#state.clients, State};

handle_call({register, ProcessLoop, Node}, _From, State) ->
	case lists:keysearch(Node, 1, State#state.clients) of
		{value, _} ->
			true;
		_ ->
			error_logger:info_msg("allocated a new resource: ~w ~w~n", [Node, ProcessLoop]),
			monitor_node(Node, true),
			link(ProcessLoop)
	end,
	{reply, {ok, node()}, State#state{clients = [{Node, ProcessLoop, 1, 0, 0} | State#state.clients]}};

handle_call(alloc, _From, State) ->
	case State#state.clients of
		[] ->
			{reply, not_allocated, State};
		Clients ->
			{Node, Client, Weight, R, Q} = random_client_pick(Clients),
			NewClients = lists:delete({Node, Client, Weight, R, Q}, Clients) ++ [{Node, Client, Weight, R + 1, Q}],
			NewState = log_request(1, State, NewClients),
			{reply, {allocated, Client}, NewState}
	end;

handle_call(Msg, _From, State) ->
	error_logger:error_msg("Invalid call: ~w~n", [Msg]),
	{noreply, State}.

%%--------------------------------------------------------------------
%% Function: handle_cast(Msg, State) -> {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, State}
%% Description: Handling cast messages
%%--------------------------------------------------------------------
handle_cast(status, State) ->
	error_logger:info_msg("Requests per client: ~w~n", [lists:map(fun({Node, _Client, Weight, R, _Q}) -> {Node, Weight, R} end, State#state.clients)]),
	error_logger:info_msg("Total requests: ~w~n", [lists:foldl(fun({_Node, _Client, _Weight, R, _Q}, Total) -> Total + R end, 0, State#state.clients)]),
	rq_server:status(),
	{noreply, State};

handle_cast({set, Node, Weight}, State) when integer(Weight) ->
	case lists:keysearch(Node, 1, State#state.clients) of
		{value, Client} ->
			{Node, Pid, _, R, Q} = Client,
			{noreply, State#state{clients = [{Node, Pid, Weight, R, Q} | lists:delete(Client, State#state.clients)]}};
		false ->
			{noreply, State}
	end;

handle_cast({dealloc, Pid}, State) ->
	case lists:keysearch(Pid, 2, State#state.clients) of
		{value, {Node, Client, Weight, R, Q}} ->
			case R > 0 of
				true ->
					{noreply, State#state{clients = lists:delete({Node, Client, Weight, R, Q}, State#state.clients) ++ [{Node, Client, Weight, R - 1, Q}]}};
				false ->
					{noreply, State}
			end;
		false ->
			{noreply, State}
	end;

handle_cast(restart_all, State) ->
	lists:foreach(fun({_Node, Client, _Weight, _R, _Q}) ->
		Client ! restart_all
	end, State#state.clients),
	{noreply, State};

handle_cast({restart, Node}, State) ->
	case lists:keysearch(Node, 1, State#state.clients) of
		{value, {Node, Client, _Weight, _R, _Q}} ->
			Client ! restart_all;
		false ->
			false
	end,
	{noreply, State};

handle_cast({client_queues, ClientQueues}, State) ->
  NewClients = utils:delete_all(nil, lists:map(fun({Node, Queue}) ->
    case lists:keysearch(Node, 1, State#state.clients) of
      {value, {Node, Pid, Weight, R, PrevQ}} ->
        case Queue of
          timeout ->
            {Node, Pid, Weight, R, PrevQ};
          Q ->
            {Node, Pid, Weight, R, Q}
        end;
      _ ->
        nil
    end
  end, ClientQueues)),
  {noreply, State#state{clients = NewClients}};

handle_cast(Msg, State) ->
	error_logger:error_msg("Invalid cast: ~w~n", [Msg]),
	{noreply, State}.

%%--------------------------------------------------------------------
%% Function: handle_info(Info, State) -> {noreply, State} |
%%                                       {noreply, State, Timeout} |
%%                                       {stop, Reason, State}
%% Description: Handling all non call/cast messages
%%--------------------------------------------------------------------
handle_info({'EXIT', Pid, _Reason}, State) -> 
	error_logger:info_msg("exited! pid went down: ~w (~w)~n", [Pid, State#state.clients]),
	case lists:keysearch(Pid, 2, State#state.clients) of
		{value, Client} ->
			{noreply, State#state{clients = lists:delete(Client, State#state.clients)}};
		false ->
			{noreply, State}
	end;

handle_info({nodedown, Node}, State) ->
	error_logger:info_msg("node went down: ~w (~w)~n", [Node, State#state.clients]),
	case lists:keysearch(Node, 1, State#state.clients) of
		{value, Client} ->
			{noreply, State#state{clients = lists:delete(Client, State#state.clients)}};
		false ->
			{noreply, State}
	end;

handle_info(Info, State) ->
	error_logger:error_msg("Invalid info: ~w~n", [Info]),
	{noreply, State}.

%%--------------------------------------------------------------------
%% Function: terminate(Reason, State) -> void()
%% Description: This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any necessary
%% cleaning up. When it returns, the gen_server terminates with Reason.
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _State) ->
	tcp_server:stop(port()),
	ok.

%%--------------------------------------------------------------------
%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
%% Description: Convert process state when code is changed
%%--------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) -> {ok, State}.

%%====================================================================
%%% Internal functions
%%====================================================================

change_logfile() ->
	case regexp:split(atom_to_list(node()), "@") of
		{ok, [Host | _]} ->
		    error_logger:logfile(close),
		    case error_logger:logfile({open, "/var/log/drproxy." ++ Host ++ ".log"}) of
				ok ->
					true;
				_ ->
					error_logger:logfile(close),
					error_logger:logfile({open, "/tmp/drproxy." ++ Host ++ ".log"})
			end,
			case file:open("/tmp/drproxy." ++ Host ++ ".pid", [write]) of
				{ok, File} ->
					file:write(File, os:getpid()),
					file:close(File);
				_ ->
					false
			end;
		_ ->
			true
	end.

alloc() ->
	gen_server:call(?MODULE, alloc).

dealloc(Pid) ->
	gen_server:cast(?MODULE, {dealloc, Pid}).

servers() ->
	drproxy:config(servers).

port() ->
	case utils:index(node(), drproxy:config(servers)) of
		{ok, N} ->
			drproxy:config(server_first_port) - 1 + N;
		no ->
			error_logger:error_msg("The server name ~w has not been specified in ~s~n", [node(), drproxy:config_file()]),
			exit({config, missing_server_specification})
	end.

start_server() ->
	Port = port(),
	Host = net_adm:localhost(),
	tcp_server:start_raw_server(Port,
		fun(Socket) -> 
			input_handler(Host, Port, Socket, "") 
		end,
		drproxy:config(server_concurrent_connections),
		0).

start_yaws() ->
	yaws:start_embedded(drproxy:config(docroot), [
		{servername, "localhost"},
		{listen, {0,0,0,0}},
		{port, drproxy:config(webport)},
		case drproxy:config(auth) of
			false ->
				{};
			User ->
			    case string:tokens(atom_to_list(User), ":") of
				[Name, Passwd] ->
			    	io:format("Starting with authentication: ~s pass ~s~n", [Name, Passwd]),
				    {authdirs, [{auth,["/"],"Auth","Basic",[{Name, Passwd}],[],false}]};
				false ->
				    io:format("invalid username password for auth--should be user:pass, instead was: ~w~n", [User]),
					{}
			    end
		end
	], [
		{debug, drproxy:config(debug)}
	]).

client_loop() ->
	receive after infinity -> true end.

random_client_pick(Clients) ->
	case length(Clients) > 1 of
		true ->
		  TotalQueue = lists:foldl(fun ({_Node, _Pid, _W, _R, Q}, Total) -> Total + Q end, 0, Clients),
		  AverageQueue = TotalQueue / length(Clients),
		  
			ChosenClients = utils:select(fun ({_Node, _Pid, _W, _R, Q}) -> Q =< AverageQueue end, Clients);
		false ->
			ChosenClients = Clients
	end,

	WeightedList = lists:flatten(lists:map(fun ({Node, Pid, Weight, R, Q}) ->
		lists:map(fun (_) -> {Node, Pid, Weight, R, Q} end, lists:seq(1,Weight))
	end, ChosenClients)),
	
	lists:nth(random:uniform(length(WeightedList)), WeightedList).

input_handler(Host, Port, Socket, Request) ->
	input_handler1(Host, Port, Socket, Request).

input_handler1(Host, Port, Socket, Request) ->
	receive
		{tcp, Socket, Bin} ->
			NewData = Request ++ binary_to_list(Bin),
			input_handler2(Host, Port, Socket, NewData);
		{tcp_closed, Socket} ->
			true
	after drproxy:config(request_timeout) ->
		tcp_server:send_data(Socket, "HTTP/1.1 500 Internal Server Error\nContent-Type: text/plain\n\nThere was an error\n\n"),
		error_logger:error_msg("could not receive request within " ++ integer_to_list(drproxy:config(request_timeout)) ++ " mili-seconds~n")
	end.

input_handler2(Host, Port, Socket, "POST" ++ Request) ->
	input_handler_with_content(Host, Port, Socket, "POST" ++ Request);
input_handler2(Host, Port, Socket, "PUT" ++ Request) ->
	input_handler_with_content(Host, Port, Socket, "PUT" ++ Request);
input_handler2(Host, Port, Socket, "GET" ++ Request) ->
	input_handler_without_content(Host, Port, Socket, "GET" ++ Request);
input_handler2(Host, Port, Socket, "OPTIONS" ++ Request) ->
	input_handler_without_content(Host, Port, Socket, "OPTIONS" ++ Request);
input_handler2(Host, Port, Socket, "HEAD" ++ Request) ->
	input_handler_without_content(Host, Port, Socket, "HEAD" ++ Request);
input_handler2(Host, Port, Socket, "DELETE" ++ Request) ->
	input_handler_without_content(Host, Port, Socket, "DELETE" ++ Request);
input_handler2(Host, Port, Socket, "TRACE" ++ Request) ->
	input_handler_without_content(Host, Port, Socket, "TRACE" ++ Request);
input_handler2(Host, Port, Socket, "CONNECT" ++ Request) ->
	input_handler_without_content(Host, Port, Socket, "CONNECT" ++ Request);
input_handler2(_Host, _Port, Socket, Request) ->
	tcp_server:send_data(Socket, "HTTP/1.1 500 Internal Server Error\nContent-Type: text/plain\n\nThere was an error\n\n"),
	error_logger:error_msg("got a bad header (not get, post, delete, put, etc.): ~s~n", [Request]).

input_handler_with_content(Host, Port, Socket, Request) ->
	case utils:split_first(Request, "\r?\n\r?\n") of
		{match, Headers, Content} ->
			case regexp:match(Headers, "[Cc]ontent-[Ll]ength: [0-9]+") of
				{match, Start, Length} ->
					ContentLength = list_to_integer(string:substr(Headers, Start + 16, Length - 16)),
					case length(Content) >= ContentLength of
						true ->
							clean_request(Host, Port, Socket, Request);
						false ->
							input_handler1(Host, Port, Socket, Request)
					end;
				nomatch ->
					clean_request(Host, Port, Socket, Request)
			end;
		nomatch ->
			input_handler1(Host, Port, Socket, Request)
	end.

input_handler_without_content(Host, Port, Socket, Request) ->
	case regexp:match(Request, "\r?\n\r?\n") of
		{match, _, _} ->
			clean_request(Host, Port, Socket, Request);
		nomatch ->
			input_handler1(Host, Port, Socket, Request)
	end.

clean_request(Host, Port, Socket, RawRequest) ->
%	Time = now(),
%	case regexp:split(Request, " ") of
%		nomatch ->
%			Uri = Request;
%		{ok, SplitRequest} when length(SplitRequest) > 2 ->
%			Uri = lists:nth(2, SplitRequest);
%        {ok, _} ->
%                Uri = Request,
%                tcp_server:send_data(Socket, "HTTP/1.1 500 Internal Server Error\nContent-Type: text/plain\n\nThere was an error\n\n"),
%                exit(normal)
%	end,
	{match, TopLine, Content} = utils:split_first(RawRequest, "\r?\n"),
	{ok, TopLineWithoutQuotes, _} = regexp:gsub(TopLine, "\"", "%22"),
	{ok, TopLineWithoutBracket, _} = regexp:gsub(TopLineWithoutQuotes, ">", "%3E"),
	{ok, Uri, _} = regexp:sub(TopLineWithoutBracket, "#.*? HTTP", " HTTP"),
	Request = Uri ++ "\r\n" ++ Content,
	Bin = list_to_binary(Request),
	request_handler(Host, Port, Socket, Uri, Request, Bin).

request_handler(Host, Port, Socket, Uri, Request, Bin) ->
	case alloc() of
		{allocated, Pid} ->
			gen_server:cast(Pid, {process, self(), Bin}),
			receive
				{response, _Client, _Resource, ResponseBin, _ResponseTime, _Attempts} ->
		            case length(binary_to_list(ResponseBin)) == 0 of
	                    true ->
							%% got a bad response, try again
							dealloc(Pid),
                            request_handler(Host, Port, Socket, Uri, Request, Bin);
	                    false ->
                            tcp_server:send_bin_data(Socket, ResponseBin)
%     		                        Timestamp = #timestamp{
%     		                                uri=Uri,
%     		                                client=Client,
%     		                                resource=Resource,
%     		                                created_at=utils:unix_time(),
%     		                                attempts=Attempts,
%     		                                response_seconds=ResponseTime,
%     		                                request_seconds=timer:now_diff(now(), Time)/1000000
%     		                        },
							%%	mnesia:transaction(fun() -> mnesia:write(Timestamp) end),
		            end;
				{failure, Client, Resource, request, timeout} ->
					error_logger:error_msg("request timeout: ~s ~w~n", [Client, Resource]),
					tcp_server:send_data(Socket, "HTTP/1.1 408 Request Timeout\nContent-Type: text/plain\n\nrequest timeout\n\n");
				{failure, Client, Resource, response, timeout} ->
					error_logger:error_msg("response timeout: ~s ~w: ~s~n", [Client, Resource, Uri]),
					tcp_server:send_data(Socket, "HTTP/1.1 408 Request Timeout\nContent-Type: text/plain\n\nresponse timeout\n\n");
				{failure, Client, Resource, request, bad_header} ->
					error_logger:error_msg("bad header: ~s ~w~n*~s*~n", [Client, Resource, Request]),
					tcp_server:send_data(Socket, "HTTP/1.1 500 Internal Server Error\nContent-Type: text/plain\n\nThere was an error\n\n");
				{failure, Client, Resource, request, enfile} ->
					error_logger:error_msg("RETRY: ran out of socket names (enfile): ~s ~w~n", [Client, Resource]),
					request_handler(Host, Port, Socket, Uri, Request, Bin);
				{failure, Client, Resource, Type, Details} ->
					error_logger:error_msg("error: ~w ~s ~w~ndetails: ~w~n", [Type, Client, Resource, Details]),
					tcp_server:send_data(Socket, "HTTP/1.1 500 Internal Server Error\nContent-Type: text/plain\n\nThere was an error\n\n")
			after 3600000 ->
				error_logger:error_msg("no response in 1 hour~n"),
				tcp_server:send_data(Socket, "HTTP/1.1 500 Internal Server Error\nContent-Type: text/plain\n\nThere was an error\n\n")
			end,
			gen_tcp:close(Socket),
			dealloc(Pid);
		not_allocated ->
			error_logger:error_msg("Error: there are no registered resources~n"),
			receive after 2000 -> request_handler(Host, Port, Socket, Uri, Request, Bin) end
	end.

log_request(R, State, Clients) ->
	[{WhenOne, OneMinute}, {WhenFive, FiveMinutes}, {WhenFifteen, FifteenMinutes}] = [State#state.one_minute, State#state.five_minutes, State#state.fifteen_minutes],
	{OldAverageOneMinute, OldAverageFiveMinutes, OldAverageFifteenMinutes} = State#state.averages,
	
	[NewOneMinute, NewAverageOneMinute] = case timer:now_diff(now(), WhenOne) / 1000000 > 60 of
		true ->
			[{now(), 0}, OneMinute / 60];
		false ->
			[{WhenOne, OneMinute + R}, OldAverageOneMinute]
	end,

	[NewFiveMinutes, NewAverageFiveMinutes] = case timer:now_diff(now(), WhenFive) / 1000000 > 300 of
		true ->
			[{now(), 0}, FiveMinutes / 300];
		false ->
			[{WhenFive, FiveMinutes + R}, OldAverageFiveMinutes]
	end,

	[NewFifteenMinutes, NewAverageFifteenMinutes] = case timer:now_diff(now(), WhenFifteen) / 1000000 > 900 of
		true ->
			[{now(), 0}, FifteenMinutes / 900];
		false ->
			[{WhenFifteen, FifteenMinutes + R}, OldAverageFifteenMinutes]
	end,

	State#state{
		clients = Clients,
		one_minute = NewOneMinute,
		five_minutes = NewFiveMinutes,
		fifteen_minutes = NewFifteenMinutes,
		averages = {NewAverageOneMinute, NewAverageFiveMinutes, NewAverageFifteenMinutes}
	}.

init() ->
	% you need all servers running before you can run init
	mnesia:stop(),
	case mnesia:create_schema(servers()) of
		{error, M} ->
			error_logger:error_msg("Error: ~w~n", [M]);
		_ ->
			ok
	end,
	mnesia:start(),
    mnesia:create_table(timestamp,
                        [{type, bag}, {attributes, record_info(fields, timestamp)}]).